package com.saga.energy.sink;

import com.saga.energy.bean.EnergyHistory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class Energy_15_fjd_history_sink extends RichSinkFunction<EnergyHistory> {

    private Connection connection = null;
    private PreparedStatement preparedStatement = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("init mysql config ... ");
        String url = "jdbc:mysql://hadoop01:3306/saga_dev?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
        connection = DriverManager.getConnection(url,"root","j5ry0#jZ7vaUt5f4");
        String insertSql = "insert into energy_15_min_fjd_history(building, func_id, meter, data_time, data_value, dt) values (?, ?, ?, ?, ?, ?)";
        preparedStatement = connection.prepareStatement(insertSql);
    }

    @Override
    public void invoke(EnergyHistory value, Context context) throws Exception {

        preparedStatement.setString(1, value.getBuilding());
        preparedStatement.setString(2, value.getFuncid());
        preparedStatement.setString(3, value.getMeter());
        preparedStatement.setString(4, value.getData_time());
        preparedStatement.setBigDecimal(5, value.getData_value());
        preparedStatement.setString(6, value.getDt());

        preparedStatement.execute();
    }

    @Override
    public void close() throws Exception {

        preparedStatement.close();
        connection.close();
    }
}
